package cn.yx.framework.configuration;

import cn.com.yx.ms.domain.mq.MqCommandType;
import cn.com.yx.ms.service.IMqService;
import cn.com.yx.ms.service.impl.MqServiceImpl;
import cn.yx.framework.log.MyLogger;
import cn.yx.framework.util.JSONUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.web.context.ContextLoader;
import org.springframework.web.socket.server.standard.SpringConfigurator;

import javax.annotation.Resource;
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.net.URI;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;

//@ServerEndpoint(value = "/ws/{ip}",configurator = SpringConfigurator.class)
@ServerEndpoint(value = "/ws/{ip}")
public class MyWebSocket {
    private static final Logger logger = LoggerFactory.getLogger(MyWebSocket.class);


    private static IMqService GLOBAL_MQSERVICE;

    public MyWebSocket() {

    }

    public MyWebSocket(IMqService iMqService) {
        GLOBAL_MQSERVICE = iMqService;
    }

    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static int onlineCount = 0;

    //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    private static CopyOnWriteArraySet<MyWebSocket> webSocketSet = new CopyOnWriteArraySet<MyWebSocket>();

    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;


    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session) throws Exception {

        if (GLOBAL_MQSERVICE == null) {
//            iMqService = ContextLoader.getCurrentWebApplicationContext().getBean(IMqService.class);
            System.out.println("iMqService is null");
        }

        MyWebSocket.onlineCount = webSocketSet.size();

        //判断是否重复连接
        MyWebSocket removeItem = null;

        for (MyWebSocket item : webSocketSet) {
            if (item.session.getPathParameters().get("ip").equals(session.getPathParameters().get("ip"))) {
                if (!item.session.getPathParameters().get("ip").contains("webpage_")) {
                    String message = "存在重复连接[" + session.getPathParameters().get("ip") + "],已被强制踢出";
                    System.out.println(message);
                    sendInfo(session.getPathParameters().get("ip"), message);
                    removeItem = item;
                    webSocketSet.remove(removeItem);  //从set中删除
                    subOnlineCount();//在线数减1
                }
            }
        }

        //session.getPathParameters().get("ip")
        this.session = session;
        webSocketSet.add(this);     //加入set中
        addOnlineCount();           //在线数加1
        System.out.println("有新连接【" + this.session.getPathParameters().get("ip") + "】加入！当前在线人数为" + getOnlineCount());
        try {
            //sendMessage(CommonConstant.CURRENT_WANGING_NUMBER.toString());
            sendMessage("有新链接加入了，当前连接数：" + getOnlineCount());
        } catch (IOException e) {
            System.out.println("IO异常");
        }

    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        if (webSocketSet.contains(this)) {
            webSocketSet.remove(this);  //从set中删除
            subOnlineCount();           //在线数减1
            System.out.println("有一连接关闭！当前在线人数为" + getOnlineCount());
        }
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        // System.out.println("来自客户端的消息:" + message);
        System.out.println("[UPCMD]客户端发送============>" + message);

        JSONObject jsonObject = null;
        String command = "";
        String devsn = "";
        boolean jsonValid = JSONUtils.isJSONValid(message);
        if (jsonValid) {

            // region << 指令 >>
            try {

                //解析content
                jsonObject = JSON.parseObject(message);
                command = jsonObject.get("command").toString();
                devsn = jsonObject.get("deviceid").toString();
                MqCommandType mqCommandType = MqCommandType.valueOf("C" + command);
                MyLogger.log("/command/ws-上行/"
                                + devsn
                                + "/"
                                + mqCommandType.getIndex()
                                + "-"
                                + mqCommandType.getName()
                                + "/",
                        message);

                GLOBAL_MQSERVICE.receive(message);

                /**
                 * 业务处理
                 * 此处特别说明：由于websocket的接收脱离springboot框架，因此无法取得spring上下文，所有服务的注入都失败
                 * 因此调用服务的方式调用业务方法
                 */

                // 异步

                // region << http 请求 >>
//                RequestConfig requestConfig = RequestConfig.custom()
//                        .setSocketTimeout(3000).setConnectTimeout(3000).build();
//                CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom()
//                        .setDefaultRequestConfig(requestConfig).build();
//
//                try {
//                    httpclient.start();
//                    final CountDownLatch latch = new CountDownLatch(1);
//                    HttpPost request = new HttpPost(BaseConstant.getServerConfig().getUrl() + "/websocket");
//
//                    List<BasicNameValuePair> urlParams = new ArrayList<>();
//                    urlParams.add(new BasicNameValuePair("message", URLEncoder.encode(message, "UTF-8")));
//                    String getUrl = EntityUtils
//                            .toString(new UrlEncodedFormEntity(urlParams));
//                    request.setURI(new URI(request.getURI().toString()
//                            + "?" + getUrl));
//                    Future<HttpResponse> future = httpclient.execute(request, new FutureCallback<HttpResponse>() {
//                        //无论完成还是失败都调用countDown()
//                        @Override
//                        public void completed(final HttpResponse response) {
//                            latch.countDown();
//                            System.out.println(request.getRequestLine() + "->"
//                                    + response.getStatusLine());
//                        }
//
//                        @Override
//                        public void failed(final Exception ex) {
//                            latch.countDown();
//                            System.out.println(request.getRequestLine() + "->" + ex);
//                        }
//
//                        @Override
//                        public void cancelled() {
//                            latch.countDown();
//                            System.out.println(request.getRequestLine()
//                                    + " cancelled");
//                        }
//                    });
//
//                    // HttpResponse response = future.get();// 获取结果
//                    // System.out.println("Response: " + response.getStatusLine());
//                    latch.await();
//                    System.out.println("Shutting down");
//                } catch (Exception e) {
//                    e.printStackTrace();
//                } finally {
//                    httpclient.close();
//                    System.out.println("close");
//                }
//                MyLogger.log("/command/ws-上行/"
//                                + devsn
//                                + "/"
//                                + mqCommandType.getIndex()
//                                + "-"
//                                + mqCommandType.getName()
//                                + "/",
//                        "http请求已发出");
                // endregion

            } catch (Exception ex) {
                ex.printStackTrace();
                logger.error(ex.getMessage());
                for (StackTraceElement stack :
                        ex.getStackTrace()) {
                    logger.error(stack.getClassName() + "|" + stack.getFileName() + "|" + stack.getMethodName() + "|" + stack.getLineNumber());
                }
            } finally {
                logger.debug("WS接收数据结束...");
            }

            // endregion
        } else {
            try {
                sendInfo(session.getPathParameters().get("ip"), "json格式错误");
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }

    }

    /**
     * 发生错误时调用
     **/
    @OnError
    public void onError(Session session, Throwable error) {
        System.out.println("发生错误");
        // error.printStackTrace();
    }


    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
        //this.session.getAsyncRemote().sendText(message);
    }


    /**
     * 群发自定义消息
     */
    public static void sendWebClientInfo(String message) throws IOException {
        for (MyWebSocket item : webSocketSet) {
            if (item.session.getPathParameters().get("ip").contains("webpage_")) {
                try {
                    item.sendMessage(message);
                } catch (IOException e) {
                    continue;
                }
            }
        }
    }

    public static void sendInfo(String ip, String message) throws Exception {


        // 下行发送日志记录
        for (MyWebSocket item : webSocketSet) {
            if (item.session.getPathParameters().get("ip").equals(ip)) {
                item.sendMessage(message);
            }
        }


    }

    public static List<String> getClientIps() throws Exception {
        List<String> ips = new ArrayList<>();
        for (MyWebSocket item : webSocketSet) {
            ips.add(item.session.getPathParameters().get("ip"));
        }
        return ips;
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        MyWebSocket.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        MyWebSocket.onlineCount--;
    }


}
